package drds.data_propagate.task;

import drds.data_propagate.binlog_event.secondary.ClientId;
import drds.data_propagate.binlog_event.secondary.Entry;
import drds.data_propagate.binlog_event.secondary.position.EntryPosition;
import drds.data_propagate.binlog_event_filter.aviater.AviaterRegexBinlogEventFilter;
import drds.data_propagate.common.AbstractLifeCycle;
import drds.data_propagate.common.DataPropagateException;
import drds.data_propagate.common.utils.JsonUtils;
import drds.data_propagate.metadata.MemoryMetaDataManager;
import drds.data_propagate.metadata.MetaDataManager;
import drds.data_propagate.parse.*;
import drds.data_propagate.parse.binlog_event_position_manager.*;
import drds.data_propagate.parse.ha.HaController;
import drds.data_propagate.parse.ha.HeartBeatHaController;
import drds.data_propagate.parse.table_meta_data.TableMetaDataClassPathXmlApplicationContextManager;
import drds.data_propagate.parse.table_meta_data.TableMetaDataFactoryImpl;
import drds.data_propagate.parse.table_meta_data.TableMetaDataStore;
import drds.data_propagate.sink.EventListSink;
import drds.data_propagate.sink.entry.EntryEventListSink;
import drds.data_propagate.sink.entry.group.GroupEventListSink;
import drds.data_propagate.store.*;
import drds.data_propagate.task.model.*;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.BooleanUtils;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class TaskImpl extends AbstractLifeCycle implements Task {
    public static final String FILTER = ".*\\\\..*";
    protected String filter = FILTER; // 过滤表达式
    protected Parameter parameters; // 对应参数

    @Setter
    @Getter
    protected Long taskIdSequense; // 和manager交互唯一标示
    @Setter
    @Getter
    protected String taskId; // 队列名字
    @Setter
    @Getter
    protected EventStore<Event> eventStore; // 有序队列
    @Setter
    @Getter
    protected EventParser eventParser; // 解析对应的数据信息
    @Setter
    @Getter
    protected EventListSink<List<Entry>> eventListSink; // 链接parse和store的桥接器
    @Setter
    @Getter
    protected MetaDataManager metaDataManager; // 消费信息管理器


    public TaskImpl(TaskConfiguration taskConfiguration) {
        this.parameters = taskConfiguration.getParameter();
        this.taskIdSequense = taskConfiguration.getId();
        this.taskId = taskConfiguration.getTaskId();
        this.filter = taskConfiguration.getFilter();

        log.info("init TaskImpl for {}-{} with parameters:{}", taskIdSequense, taskId, parameters);

        // 初始化metaManager
        initMetaDataManager();
        // 初始化eventStore
        initEventStore();
        // 初始化eventSink
        initEventSink();
        // 初始化eventParser;
        initEventParser();

        // 基础工具，需要提前start，会有先订阅再根据filter条件启动parse的需求

        if (!metaDataManager.isStart()) {
            metaDataManager.start();
        }
        log.info("init successful....");
    }


    //启动的时候组件逆向启动,停止的时候顺序关闭
    @Override
    public void start() {
        super.start();
        // 初始化metaManager
        log.info("start CannalInstance for {}-{} with parameters:{}", taskIdSequense, taskId, parameters);
        if (!metaDataManager.isStart()) {
            metaDataManager.start();
        }

        if (!eventStore.isStart()) {
            eventStore.start();
        }

        if (!eventListSink.isStart()) {
            eventListSink.start();
        }

        if (!eventParser.isStart()) {
            beforeEventParserStart(eventParser);
            eventParser.start();
            afterEventParserStart(eventParser);
        }
        log.info("start successful....");
    }

    @Override
    public void stop() {
        super.stop();
        log.info("stop Task for {}-{} ", new Object[]{taskIdSequense, taskId});

        if (eventParser.isStart()) {
            beforeEventParserStop(eventParser);
            eventParser.stop();
            afterEventParserStop(eventParser);
        }

        if (eventListSink.isStart()) {
            eventListSink.stop();
        }

        if (eventStore.isStart()) {
            eventStore.stop();
        }

        if (metaDataManager.isStart()) {
            metaDataManager.stop();
        }

        log.info("stop successful....");
    }

    protected void beforeEventParserStart(EventParser eventParser) {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup) {
            // 处理group的模式
            List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (EventParser eventParser1 : eventParserList) {// 需要遍历启动
                startEventParserInternal(eventParser1, true);
            }
        } else {
            startEventParserInternal(eventParser, false);
        }
    }

    /**
     * 初始化单个eventParser，不需要考虑group
     */
    protected void startEventParserInternal(EventParser eventParser, boolean isGroup) {
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            BinLogEventPositionManager binLogEventPositionManager = abstractEventParser.getBinLogEventPositionManager();
            if (!binLogEventPositionManager.isStart()) {
                binLogEventPositionManager.start();
            }
        }

        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            HaController haController = mysqlEventParser.getHaController();

            if (haController instanceof HeartBeatHaController) {
                ((HeartBeatHaController) haController).setHaSwitchable(mysqlEventParser);
            }

            if (!haController.isStart()) {
                haController.start();
            }

        }
    }

    // around protogenesis parser, default impl
    protected void afterEventParserStart(EventParser eventParser) {
        // 读取一下历史订阅的filter信息
        List<ClientId> clientIdList = metaDataManager.listAllSubscribeInfo(taskId);
        for (ClientId clientId : clientIdList) {
            subscribe(clientId);
        }
    }

    public boolean subscribe(ClientId identity) {
        if (StringUtils.isNotEmpty(identity.getFilter())) {
            log.info("subscribe binlogEventFilter change decode " + identity.getFilter());
            AviaterRegexBinlogEventFilter aviaterRegexFilter = new AviaterRegexBinlogEventFilter(identity.getFilter());

            boolean isGroup = (eventParser instanceof GroupEventParser);
            if (isGroup) {
                // 处理group的模式
                List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
                for (EventParser eventParser : eventParserList) {// 需要遍历启动
                    ((AbstractEventParser) eventParser).setBinlogEventFilter(aviaterRegexFilter);
                }
            } else {
                ((AbstractEventParser) eventParser).setBinlogEventFilter(aviaterRegexFilter);
            }

        }

        // filter的处理规则
        // a. parser处理数据过滤处理
        // b. sink处理数据的路由&分发,一份parse数据经过sink后可以分发为多份，每份的数据可以根据自己的过滤规则不同而有不同的数据
        // 后续内存版的一对多分发，可以考虑
        return true;
    }

    // around protogenesis parser
    protected void beforeEventParserStop(EventParser eventParser) {
        // noop
    }

    protected void afterEventParserStop(EventParser eventParser) {

        boolean isGroup = (eventParser instanceof GroupEventParser);
        if (isGroup) {
            // 处理group的模式
            List<EventParser> eventParserList = ((GroupEventParser) eventParser).getEventParserList();
            for (EventParser eventParser1 : eventParserList) {// 需要遍历启动
                stopEventParserInternal(eventParser1);
            }
        } else {
            stopEventParserInternal(eventParser);
        }
    }


    protected void stopEventParserInternal(EventParser eventParser) {
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            // 首先启动log position管理器
            BinLogEventPositionManager binLogEventPositionManager = abstractEventParser.getBinLogEventPositionManager();
            if (binLogEventPositionManager.isStart()) {
                binLogEventPositionManager.stop();
            }
        }

        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;
            HaController haController = mysqlEventParser.getHaController();
            if (haController.isStart()) {
                haController.stop();
            }
        }
    }


    protected void initMetaDataManager() {
        log.info("init metaDataManager begin...");
        metaDataManager = new MemoryMetaDataManager();

        log.info("init metaDataManager end! \n\t load MetaDataManager:{} ", metaDataManager.getClass().getName());
    }

    protected void initEventStore() {
        log.info("init eventStore begin...");
        MemoryEventStore memoryEventStore = new MemoryEventStore(BatchMode.item_size);//默认是item_size
        memoryEventStore.setBufferSize(parameters.getMemoryStorageBufferSize());
        memoryEventStore.setBufferMemUnit(parameters.getMemoryStorageBufferMemUnit());
        memoryEventStore.setBatchMode(BatchMode.item_size);
        memoryEventStore.setDdlIsolation(parameters.getDdlIsolation());
        memoryEventStore.setRaw(parameters.getMemoryStorageRawEntry());
        eventStore = memoryEventStore;

        if (eventStore instanceof AbstractStoreScavenge) {
            StorageScavengeMode scavengeMode = parameters.getStorageScavengeMode();
            AbstractStoreScavenge eventScavengeStore = (AbstractStoreScavenge) eventStore;
            eventScavengeStore.setTaskId(taskId);
            eventScavengeStore.setMetaDataManager(metaDataManager);
            eventScavengeStore.setOnAck(scavengeMode.isOnAck());
            eventScavengeStore.setOnFull(scavengeMode.isOnFull());
            eventScavengeStore.setOnSchedule(scavengeMode.isOnSchedule());
            if (scavengeMode.isOnSchedule()) {
                eventScavengeStore.setScavengeSchedule(parameters.getScavengeSchdule());
            }
        }
        memoryEventStore.start();
        log.info("init eventStore end! \n\t load EventStore:{}", eventStore.getClass().getName());
    }

    protected void initEventSink() {
        log.info("init eventListSink begin...");

        int groupSize = getGroupSize();
        if (groupSize <= 1) {
            eventListSink = new EntryEventListSink();
        } else {
            eventListSink = new GroupEventListSink(groupSize);
        }

        if (eventListSink instanceof EntryEventListSink) {
            ((EntryEventListSink) eventListSink).setNotUseTransactionTimelineBarrier(false);
            ((EntryEventListSink) eventListSink).setEventStore(getEventStore());
        }
        log.info("init eventListSink end! \n\t load EventListSink:{}", eventListSink.getClass().getName());
    }

    protected void initEventParser() {
        log.info("init eventParser begin...");
        BinlogEventParserType binlogEventParserType = parameters.getBinlogEventParserType();

        List<List<DataSourcing>> groupDbAddresses = parameters.getGroupDbAddresses();
        if (!CollectionUtils.isEmpty(groupDbAddresses)) {
            int size = groupDbAddresses.get(0).size();// 取第一个分组的数量，主备分组的数量必须一致
            List<EventParser> eventParserList = new ArrayList<EventParser>();
            for (int i = 0; i < size; i++) {
                List<InetSocketAddress> dbAddress = new ArrayList<InetSocketAddress>();
                BinlogEventParserType lastType = null;
                for (List<DataSourcing> groupDbAddress : groupDbAddresses) {
                    if (lastType != null && !lastType.equals(groupDbAddress.get(i).getBinlogEventParserType())) {
                        throw new DataPropagateException(
                                String.format("master/slave Sourcing eventType is unmatch. %s vs %s", lastType,
                                        groupDbAddress.get(i).getBinlogEventParserType()));
                    }

                    lastType = groupDbAddress.get(i).getBinlogEventParserType();
                    dbAddress.add(groupDbAddress.get(i).getDbAddress());
                }

                // 初始化其中的一个分组parser
                eventParserList.add(doInitEventParser(lastType, dbAddress));
            }

            if (eventParserList.size() > 1) { // 如果存在分组，构造分组的parser
                GroupEventParser groupEventParser = new GroupEventParser();
                groupEventParser.setEventParserList(eventParserList);
                this.eventParser = groupEventParser;
            } else {
                this.eventParser = eventParserList.get(0);
            }
        } else {
            // 创建一个空数据库地址的parser，可能使用了tddl指定地址，启动的时候才会从tddl获取地址
            this.eventParser = doInitEventParser(binlogEventParserType, new ArrayList<InetSocketAddress>());
        }

        log.info("init eventParser end! \n\t load EventParser:{}", eventParser.getClass().getName());
    }

    private EventParser doInitEventParser(BinlogEventParserType binlogEventParserType, List<InetSocketAddress> inetSocketAddressList) {
        EventParser eventParser = null;
        if (BinlogEventParserType.db.equals(binlogEventParserType)) {
            MysqlEventParser mysqlEventParser = new MysqlEventParser();
            mysqlEventParser.setTaskId(taskId);
            // 编码参数
            mysqlEventParser.setConnectionCharset(Charset.forName(parameters.getConnectionCharset()));
            mysqlEventParser.setConnectionCharsetNumber(parameters.getConnectionCharsetNumber());
            // 网络相关参数
            mysqlEventParser.setDefaultConnectionTimeoutInSeconds(parameters.getDefaultConnectionTimeoutInSeconds());
            mysqlEventParser.setSendBufferSize(parameters.getSendBufferSize());
            mysqlEventParser.setReceiveBufferSize(parameters.getReceiveBufferSize());
            // 心跳检查参数
            mysqlEventParser.setHeartBeatEnable(parameters.getDetectingEnable());
            mysqlEventParser.setHeartBeatSql(parameters.getDetectingSQL());
            mysqlEventParser.setDetectingIntervalInSeconds(parameters.getDetectingIntervalInSeconds());
            // 数据库信息参数
            mysqlEventParser.setSlaveId(parameters.getSlaveId());
            if (!CollectionUtils.isEmpty(inetSocketAddressList)) {
                mysqlEventParser.setMasterAuthenticationInfo(new AuthenticationInfo(inetSocketAddressList.get(0), parameters.getDbUsername(),
                        parameters.getDbPassword(), parameters.getDefaultDatabaseName()));

                if (inetSocketAddressList.size() > 1) {
                    mysqlEventParser
                            .setStandbyAuthenticationInfo(new AuthenticationInfo(inetSocketAddressList.get(1), parameters.getDbUsername(),
                                    parameters.getDbPassword(), parameters.getDefaultDatabaseName()));
                }
            }

            if (!CollectionUtils.isEmpty(parameters.getPositions())) {
                EntryPosition masterPosition = JsonUtils.unmarshalFromString(parameters.getPositions().get(0),
                        EntryPosition.class);
                // binlog位置参数
                mysqlEventParser.setMasterEntryPosition(masterPosition);

                if (parameters.getPositions().size() > 1) {
                    EntryPosition standbyPosition = JsonUtils.unmarshalFromString(parameters.getPositions().get(1),
                            EntryPosition.class);
                    mysqlEventParser.setStandbyEntryPosition(standbyPosition);
                }
            }
            mysqlEventParser.setFallbackIntervalInSeconds(parameters.getFallbackIntervalInSeconds());
            //mysqlEventParser.getProfilingEnabled().set(false);
            mysqlEventParser.setFilterTableError(parameters.getFilterTableError());
            mysqlEventParser.setParallel(parameters.getParallel());
            mysqlEventParser.setGtidMode(BooleanUtils.toBoolean(parameters.getGtidEnable()));
            // table_meta_data
            if (parameters.getTsdbSnapshotInterval() != null) {
                mysqlEventParser.setTsdbSnapshotInterval(parameters.getTsdbSnapshotInterval());
            }
            if (parameters.getTsdbSnapshotExpire() != null) {
                mysqlEventParser.setTsdbSnapshotExpire(parameters.getTsdbSnapshotExpire());
            }
            boolean tsdbEnable = BooleanUtils.toBoolean(parameters.getTsdbEnable());
            if (tsdbEnable) {
                mysqlEventParser.setTableMetaDataFactory(new TableMetaDataFactoryImpl() {

                    @Override
                    public void destory(String taskId) {
                        TableMetaDataClassPathXmlApplicationContextManager.destory(taskId);
                    }

                    @Override
                    public TableMetaDataStore build(String taskId, String springXml) {
                        try {
                            System.setProperty("canal.task.tsdb.url", parameters.getTsdbJdbcUrl());
                            System.setProperty("canal.task.tsdb.dbUsername", parameters.getTsdbJdbcUserName());
                            System.setProperty("canal.task.tsdb.dbPassword", parameters.getTsdbJdbcPassword());

                            return TableMetaDataClassPathXmlApplicationContextManager.build(taskId,
                                    "classpath:spring/table_meta_data/mysql-table_meta_data.xml");
                        } finally {
                            System.setProperty("canal.task.tsdb.url", "");
                            System.setProperty("canal.task.tsdb.dbUsername", "");
                            System.setProperty("canal.task.tsdb.dbPassword", "");
                        }
                    }
                });
                mysqlEventParser.setEnableTableMetaDataCache(tsdbEnable);
            }
            eventParser = mysqlEventParser;
        }

        // add transaction authentication_info at 2012-12-06
        if (eventParser instanceof AbstractEventParser) {
            AbstractEventParser abstractEventParser = (AbstractEventParser) eventParser;
            abstractEventParser.setEventListqQueueSize(parameters.getTransactionSize());
            abstractEventParser.setBinLogEventPositionManager(initLogPositionManager());

            abstractEventParser.setEventListSink(this.getEventListSink());

            if (StringUtils.isNotEmpty(filter)) {
                AviaterRegexBinlogEventFilter aviaterRegexFilter = new AviaterRegexBinlogEventFilter(filter);
                abstractEventParser.setBinlogEventFilter(aviaterRegexFilter);
            }

            // 设置黑名单
            if (StringUtils.isNotEmpty(parameters.getBlackFilter())) {
                AviaterRegexBinlogEventFilter aviaterRegexFilter = new AviaterRegexBinlogEventFilter(parameters.getBlackFilter());
                abstractEventParser.setEventBlackFilter(aviaterRegexFilter);
            }
        }
        if (eventParser instanceof MysqlEventParser) {
            MysqlEventParser mysqlEventParser = (MysqlEventParser) eventParser;

            // 初始化haController，绑定与eventParser的关系，haController会控制eventParser
            HaController haController = initHaController();
            mysqlEventParser.setHaController(haController);
        }
        return eventParser;
    }

    protected HaController initHaController() {
        log.info("init haController begin...");

        HaController haController = new HeartBeatHaController();
        ((HeartBeatHaController) haController).setDetectingRetryTimes(parameters.getDetectingRetryTimes());
        ((HeartBeatHaController) haController).setSwitchEnable(parameters.getHeartbeatHaEnable());
        log.info("init haController end! \n\t load HaController:{}", haController.getClass().getName());

        return haController;
    }

    protected BinLogEventPositionManager initLogPositionManager() {
        log.info("init logPositionPersistManager begin...");
        IndexMode indexMode = parameters.getIndexMode();
        BinLogEventPositionManager binLogEventPositionManager;
        if (indexMode.isMemory()) {
            binLogEventPositionManager = new MemoryBinLogEventPositionManager();
        } else if (indexMode.isZookeeper()) {
            binLogEventPositionManager = new ConfigurationServerBinLogEventPositionManager();
        } else if (indexMode.isMixed()) {
            MemoryBinLogEventPositionManager memoryLogPositionManager = new MemoryBinLogEventPositionManager();
            ConfigurationServerBinLogEventPositionManager zooKeeperLogPositionManager = new ConfigurationServerBinLogEventPositionManager();
            binLogEventPositionManager = new PeriodMixedBinLogEventPositionManager(memoryLogPositionManager,
                    zooKeeperLogPositionManager, 1000L);
        } else if (indexMode.isMeta()) {
            binLogEventPositionManager = new MetaBinLogEventPositionManager(metaDataManager);
        } else if (indexMode.isMemoryMetaFailback()) {
            MemoryBinLogEventPositionManager primary = new MemoryBinLogEventPositionManager();
            MetaBinLogEventPositionManager secondary = new MetaBinLogEventPositionManager(metaDataManager);

            binLogEventPositionManager = new FailbackBinLogEventPositionManager(primary, secondary);
        } else {
            throw new DataPropagateException("unsupport indexMode for " + indexMode);
        }

        log.info("init binLogEventPositionManager end! \n\t load BinLogEventPositionManager:{}",
                binLogEventPositionManager.getClass().getName());

        return binLogEventPositionManager;
    }


    private int getGroupSize() {
        List<List<DataSourcing>> groupDbAddresses = parameters.getGroupDbAddresses();
        if (!CollectionUtils.isEmpty(groupDbAddresses)) {
            return groupDbAddresses.get(0).size();
        } else {
            // 可能是基于tddl的启动
            return 1;
        }
    }

}
